package com.sunl.mall.ali.mq.producer;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import com.sunl.mall.ali.mq.api.producer.RocketMqOrderProducer;
import com.sunl.mall.ali.mq.model.RocketMqSendResult;
import com.sunl.mall.ali.mq.model.order.RocketMqOrderMessageWrapper;
import com.sunl.mall.ali.mq.serializer.RocketMqSerializer;
import com.sunl.mall.common.api.util.ExceptionUtil;
import com.sunl.mall.common.api.util.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 对于顺序消息的__SHARDINGKEY字段有以下解释：
 * 分区顺序消息中区分不同分区的关键字段，Sharding Key与普通消息的key是完全不同的概念。
 * 全局顺序消息，该字段可以设置为任意非空字符串。
 */
public class RocketMqOrderProducerBean extends AbstractRocketMqProducer implements RocketMqOrderProducer {
    private final static Logger log = LoggerFactory.getLogger(RocketMqProducerBean.class);

    /**
     * 需要注入该字段，指定构造{@link OrderProducer}实例的属性，具体支持的属性详见{@link PropertyKeyConst}
     */
    private final Properties properties;

    private OrderProducer orderProducer;

    public RocketMqOrderProducerBean(Properties properties) {
        this.properties = properties;
    }

    @Override
    public <T> boolean send(RocketMqOrderMessageWrapper<T> message, RocketMqSerializer<T> serializer) {
        return sendBackResult(message, serializer).isSuccess();
    }

    @Override
    public <T> RocketMqSendResult sendBackResult(RocketMqOrderMessageWrapper<T> message, RocketMqSerializer<T> serializer) {
        if (message.getContent() == null) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey()};
            log.error("Mq send Failure, Because Message content is null , topic:[{}], tag:[{}], key:[{}]", params);
            return RocketMqSendResult.fail();
        }
        SendResult sendResult = null;
        long start = SystemClock.currentTimeMillis();
        try {
            byte[] bytes = serializer.serialize(message.getContent());
            message.setBody(bytes);
            if (!checkSendMsgBody(message)) {
                return RocketMqSendResult.fail();
            }
            sendResult = this.orderProducer.send(message, message.getShardingKey());
        } catch (Throwable e) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey(), message.getContent(), message.getShardingKey(), SystemClock.currentTimeMillis() - start, ExceptionUtil.getAsString(e)};
            log.error("Mq send order message Exception, topic:[{}], tag:[{}], key:[{}], message:[{}], shardingKey[{}], costTime:{}ms, Some Exception Occur:[{}]", params);
        }
        return RocketMqSendResult.successIfNotNull(sendResult);
    }

    @Override
    public boolean isStarted() {
        return this.orderProducer.isStarted();
    }

    @Override
    public boolean isClosed() {
        return this.orderProducer.isClosed();
    }

    @Override
    public void start() {
        if (null == this.properties) {
            throw new ONSClientException("properties not set");
        }
        init(this.properties);
        this.orderProducer = ONSFactory.createOrderProducer(this.properties);
        this.orderProducer.start();
    }

    @Override
    public void updateCredential(Properties credentialProperties) {
        if (this.orderProducer != null) {
            this.orderProducer.updateCredential(credentialProperties);
        }
    }

    @Override
    public void shutdown() {
        if (this.orderProducer != null) {
            this.orderProducer.shutdown();
        }
    }
}
